unit Actors;

{$IFDEF FPC}
  {$MODE Delphi}{$H+}
{$ENDIF}

interface

{.$DEFINE PARALLELJOBS}

uses
  Windows, Classes, Variants
{$IFDEF PARALLELJOBS}
  , ParallelJobs
{$ENDIF}
  ;

type
  PActParam = ^PPointer;

  TAct = procedure (AParam: PActParam) of object;

  TPooledActors = class;

{$IFDEF PARALLELJOBS}
  TPoolThread = class
  private
    FParallelJob: TParallelJob;  
{$ELSE}
  TPoolThread = class(TThread)
  private
{$ENDIF}
    FActived: boolean;
    FActsCount: integer;
    FPooling: THandle;
    FPool: TPooledActors;
    procedure SetActive(const Value: boolean);
  public
    constructor Create(APool: TPooledActors);
    {$IFDEF PARALLELJOBS}
    destructor Destroy; override;
    procedure Execute;
    {$ELSE}
    procedure Execute; override;
    {$ENDIF}
    procedure StopThread;
  
    property Active: boolean read FActived write SetActive;
  end;

  PActor = ^TActor;  
  TActor = class;

  PActData = ^TActData;
  TActData = packed record
    Act: TAct;
    Data: PActParam;
    Dest: TActor;
    Sender: TActor;
    Next: PActData;
  end;        

  TActList = packed record
    F, L: PActData;
  end;

  TActorList = packed record
    F, L: TActor;
  end;

  TPooledActors = class
  private
    FThreads: TThreadList;
    FActors: TActorList;
    FThreadsCount: integer;
    FThreadsActive: integer;
    FActorHasAct: THandle;
    FCS: TRTLCriticalSection;
    FSysInfo: TSystemInfo;

    FAutoBalance: boolean;
    FActingCounter: Integer;
    FActingMaxCounter: Integer;

    procedure InitThreads(AThreadsCount: integer);
    procedure SetThreadsCount(const Value: Integer);
    procedure ActiveThreads(AThreadsCount: integer);

    procedure ActingStart(var AActData: PActData); virtual;
    procedure ActingEnd(var AActData: PActData); virtual;
  public
    constructor Create; overload;
    constructor Create(AThreadsCount: integer); overload;
    destructor Destroy; override;

    function AddActor(AActor: TActor): TActor; virtual;
    procedure RemoveActor(AActor: TActor); virtual;

    procedure Dispatch(APoolThread: TPoolThread); virtual;

    property ThreadsCount: Integer read FThreadsCount write SetThreadsCount;
    property ActingMaxCounter: integer read FActingMaxCounter write FActingMaxCounter;
    property ThreadsActive: integer read FThreadsActive write FThreadsActive;
  end;

  TActor = class
  private
    FName: string;
    FActs: TActList;
    FPool: TPooledActors;
    FActLock: integer;
    FCS: TRTLCriticalSection;
    FNext: TActor;
    FPrior: TActor;    
  public
    constructor Create(const AName: string);
    destructor Destroy; override;

    procedure Loaded; virtual;

    function Act: boolean; virtual;

    procedure Queue(ASender: TActor; var AActData: PActData);
    function Pop(var AActData: PActData): boolean;

    function HasAct: Boolean;

    procedure Send(ADestActor: TActor; AAct: TAct; AParam: Pointer); overload;
    procedure Send(ADestActor: TActor; AAct: TAct; AParam: array of const); overload;

    procedure Send(AAct: TAct; AParam: Pointer); overload;
    procedure Send(AAct: TAct; AParam: array of const); overload;

    property Name: string read FName;
  end;

implementation

uses Math;

function ICEInteger(var D: integer; C, E: integer) : boolean;
{$IFDEF VER150}
  var
    O: Pointer;
  begin
    O := InterlockedCompareExchange(Pointer(D), Pointer(E), Pointer(C));
    Result := Integer(O) = C;
  end;
{$ELSE}
  begin
    Result := (InterlockedCompareExchange(D,E,C) = C);
  end;
{$ENDIF}

function ActParamToBuffer(const AParam: array of const): Pointer;
var
  WriteAt: Cardinal;
  i, NeededSize: integer;
begin
  NeededSize := 0;
  for I := 0 to High(AParam) do
  begin
    case AParam[i].VType of
      vtBoolean, vtChar: Inc(NeededSize, 1);

      vtInteger, vtString, vtPChar, vtObject, vtClass,
      vtWideChar, vtPWideChar, vtAnsiString, vtWideString,
      vtVariant, vtInterface: Inc(NeededSize, 4);

      vtCurrency, vtExtended, vtInt64: Inc(NeededSize, 8);
    end;
  end;

  GetMem(Result, NeededSize);
  WriteAt := 0;
  for I := 0 to High(AParam) do
  begin
    case AParam[i].VType of
      vtBoolean, vtChar: begin
        Move(AParam[i], Pointer(Cardinal(Result) + WriteAt)^, 1);
        Inc(WriteAt, 1);
      end;

      vtInteger, vtString, vtPChar, vtObject, vtClass,
      vtWideChar, vtPWideChar, vtAnsiString, vtWideString,
      vtVariant, vtInterface: begin
        Move(AParam[i], Pointer(Cardinal(Result) + WriteAt)^, 4);
        Inc(WriteAt, 4);
      end;

      vtCurrency, vtExtended, vtInt64: begin
        Move(AParam[i], Pointer(Cardinal(Result) + WriteAt)^, 8);
        Inc(WriteAt, 8);
      end;
    end;
  end;
end;

type
  PDataParam = ^TDataParam;
  TDataParam = packed record
    refCount: Integer;
    data: Pointer;
  end;

{ TPoolThread }

constructor TPoolThread.Create(APool: TPooledActors);
begin
  FPool := APool;

  FActived := false;
  FPooling := CreateEvent(nil, true, false, nil);

  FActsCount := 0;

{$IFDEF PARALLELJOBS}
  FParallelJob := TParallelJob.Create(Self, @TPoolThread.Execute);
  FParallelJob.Start;
{$ELSE}
  FreeOnTerminate := false;
  inherited Create(false);
{$ENDIF}
end;

{$IFDEF PARALLELJOBS}
destructor TPoolThread.Destroy;
begin
  FParallelJob.Free;

  inherited;
end;
{$ENDIF}

procedure TPoolThread.Execute;
begin
{$IFDEF PARALLELJOBS}
  while not FParallelJob.Info.Terminated do
{$ELSE}
  while not Terminated do
{$ENDIF}
  begin
    WaitForSingleObject(FPooling, INFINITE);
    FPool.Dispatch(Self);
  end;
end;

procedure TPoolThread.StopThread;
begin
{$IFDEF PARALLELJOBS}
  FParallelJob.Stop(True);
{$ELSE}
  Terminate;
  WaitFor;
{$ENDIF}
end;

procedure TPoolThread.SetActive(const Value: boolean);
begin
  FActived := Value;
  if FActived then
    SetEvent(FPooling)
  else
    ResetEvent(FPooling);
end;

{ TPooledActors }

constructor TPooledActors.Create;
begin
  GetSystemInfo(FSysInfo);
  FThreads := TThreadList.Create;

  InitializeCriticalSection(FCS);
  InitThreads(FSysInfo.dwNumberOfProcessors);

  FActingCounter := 0;
  FAutoBalance := true;
  ActiveThreads(1);
end;

constructor TPooledActors.Create(AThreadsCount: integer);
begin
  GetSystemInfo(FSysInfo);
  FThreads := TThreadList.Create;

  InitializeCriticalSection(FCS);
  InitThreads(AThreadsCount);

  FActingCounter := 0;
  FAutoBalance := false;
  ActiveThreads(AThreadsCount);
end;

destructor TPooledActors.Destroy;
begin
  InitThreads(0);

  FThreads.Free;

  inherited;
end;

function TPooledActors.AddActor(AActor: TActor): TActor;
begin
  EnterCriticalSection(FCS);
  try
    AActor.FPool := Self;
    if FActors.F = nil then
    begin
      FActors.F := AActor;
      FActors.L := AActor;      
    end else
    begin
      FActors.L.FNext := AActor;
      AActor.FPrior := FActors.L;
      FActors.L := AActor;
    end;
  finally
    LeaveCriticalSection(FCS);
  end;
end;

procedure TPooledActors.InitThreads(AThreadsCount: integer);
var
  i: integer;
begin
  with FThreads.LockList do
    try
      for i := 0 to Count - 1 do
      begin
        TPoolThread(Items[i]).Active := true;
        TPoolThread(Items[i]).StopThread;
        TPoolThread(Items[i]).Free;
      end;

      FThreads.Clear;

      for i := 0 to AThreadsCount - 1 do
        Add(TPoolThread.Create(Self));

      FThreadsCount := AThreadsCount;
    finally
      FThreads.UnlockList;
    end;
end;

procedure TPooledActors.Dispatch(APoolThread: TPoolThread);
var
  i: integer;
  bAct: boolean;
  CurrActor, CurrActorPrior: TActor;
begin
  bAct := false;
  EnterCriticalSection(FCS);
  try
    CurrActor := FActors.F;
    while CurrActor <> nil do
    begin
      if CurrActor.HasAct then
        Break;
      CurrActor := CurrActor.FNext;
    end;

    if CurrActor <> nil then
    begin
      CurrActorPrior := CurrActor.FPrior;
      CurrActor.FPrior := CurrActor.FNext;
    end;
  finally
    LeaveCriticalSection(FCS);
  end;

  if CurrActor <> nil then
  begin
    if CurrActor.Act then
      bAct := true;

    if not bAct then
      Sleep(1)
    else
      Inc(APoolThread.FActsCount, 1);

    EnterCriticalSection(FCS);
    try
      CurrActor.FPrior := CurrActorPrior;
    finally
      LeaveCriticalSection(FCS);
    end;
  end;
end;

procedure TPooledActors.SetThreadsCount(const Value: Integer);
begin
  if FThreadsCount <> Value then
  begin
    InitThreads(Value);
    ActiveThreads(Value);
  end;
end;

procedure TPooledActors.RemoveActor(AActor: TActor);
var
  CurrActor: TActor;
begin
  EnterCriticalSection(FCS);
  try
    CurrActor := FActors.F;
    while CurrActor <> nil do
    begin
      if CurrActor = AActor then
      begin
        if CurrActor = FActors.L then
          FActors.L := FActors.L.FPrior;
        if CurrActor = FActors.F then
          FActors.F := FActors.F.FNext;
        CurrActor.FPrior := CurrActor.FNext;
        Break;
      end;

      CurrActor := CurrActor.FNext;
    end;
  finally
    LeaveCriticalSection(FCS);
  end;
end;

procedure TPooledActors.ActiveThreads(AThreadsCount: integer);
var
  i, x, ProcessorRange: integer;
  AffinityMask: DWORD;
  PoolThread: TPoolThread;
begin
  with FThreads.LockList do
    try
      ProcessorRange := FSysInfo.dwNumberOfProcessors div AThreadsCount;
      if ProcessorRange = 0 then
        ProcessorRange := 1;

      FThreadsActive := AThreadsCount;
      for i := 0 to Count - 1 do
      begin
        PoolThread := TPoolThread(Items[i]);
        PoolThread.Active := (i + 1) <= AThreadsCount;

        if PoolThread.Active then
        begin
          if ProcessorRange = 1 then
          begin
            AffinityMask := 1;
            if i > 0 then
              AffinityMask := 1 shl i;
          end else
          begin
            AffinityMask := 0;
            for x := 1 to ProcessorRange do
              AffinityMask := (AffinityMask shl 1) + (1 shl i);
          end;

          {$IFDEF PARALLELJOBS}
            SetThreadAffinityMask(PoolThread.FParallelJob.Info.Handle, AffinityMask);
          {$ELSE}
            SetThreadAffinityMask(PoolThread.Handle, AffinityMask);
          {$ENDIF}    
        end;
      end;
    finally
      FThreads.UnlockList;
    end;
end;

procedure TPooledActors.ActingStart(var AActData: PActData);
begin
  if FAutoBalance then
  begin
    EnterCriticalSection(FCS);
    try
      InterlockedIncrement(FActingCounter);
      if FActingCounter > FActingMaxCounter then
      begin
        InterlockedExchange(FActingMaxCounter, FActingCounter);
        if FActingMaxCounter > FSysInfo.dwNumberOfProcessors then
          ActiveThreads(FSysInfo.dwNumberOfProcessors)
        else if FActingMaxCounter > 1 then
          ActiveThreads(FActingMaxCounter)
        else
          ActiveThreads(1);        
      end;
    finally
      LeaveCriticalSection(FCS);
    end;
  end;
end;

procedure TPooledActors.ActingEnd(var AActData: PActData);
begin
  if FAutoBalance then
  begin
    EnterCriticalSection(FCS);
    try
      InterlockedDecrement(FActingCounter);
      if FActingCounter = 0 then
        InterlockedExchange(FActingMaxCounter, FActingCounter);
    finally
      LeaveCriticalSection(FCS);
    end;
  end;
end;

{ TActor }

constructor TActor.Create(const AName: string);
begin
  FName := AName;
  FActLock := 1;
  InitializeCriticalSection(FCS);
  
  FActs.F := nil;
  FActs.L := nil;

  FNext := nil;
  FPrior := nil;

  Loaded;
end;

destructor TActor.Destroy;
begin
  if FPool <> nil then
    FPool.RemoveActor(Self);

  inherited;
end;

procedure TActor.Loaded;
begin
  //
end;

function TActor.Act: boolean;
var
  CurrAct: PActData;
  Data: PDataParam;
begin
  Result := false;
  while Pop(CurrAct) do
  begin
    CurrAct.Act(CurrAct.Data);
    FPool.ActingEnd(CurrAct);

    Data := PDataParam(Cardinal(CurrAct.Data) - 4);
    InterlockedDecrement(Data^.refCount);
    if Data^.refCount = 0 then
    begin
      FreeMem(Data^.data);
      Dispose(Data);
    end;

    Dispose(CurrAct);
    Result := true;
  end;
end;

function TActor.HasAct: Boolean;
begin
  Result := ICEInteger(FActLock, 0, 0);
end;

procedure TActor.Queue(ASender: TActor; var AActData: PActData);
begin
  EnterCriticalSection(FCS);
  try
    if FActs.F = nil then
      FActs.F := AActData
    else
      FActs.L.Next := AActData;
    FActs.L := AActData;
  finally
    LeaveCriticalSection(FCS);
  end;

  ICEInteger(FActLock, 1, 0);
  FPool.ActingStart(AActData);
end;

function TActor.Pop(var AActData: PActData): boolean;
begin
  Result := false;
  if ICEInteger(FActLock, 0, 1) then
  begin
    EnterCriticalSection(FCS);
    try
      if FActs.F <> nil then
      begin
        AActData := PActData(FActs.F);
        FActs.F := FActs.F.Next;
        if FActs.F <> nil then
          ICEInteger(FActLock, 1, 0);
        Result := True;            
      end;
    finally
      LeaveCriticalSection(FCS);
    end;
  end;
end;

procedure TActor.Send(ADestActor: TActor; AAct: TAct; AParam: Pointer);
var
  NewAct: PActData;
  data: PDataParam;  
begin
  New(NewAct);
  NewAct.Act := AAct;
  NewAct.Dest := ADestActor;
  NewAct.Sender := Self;
  NewAct.Data := AParam;
  NewAct.Next := nil;

  Data := PDataParam(Cardinal(AParam) - 4);
  InterlockedIncrement(Data^.refCount);  

  ADestActor.Queue(Self, NewAct);
end;

procedure TActor.Send(ADestActor: TActor; AAct: TAct;
  AParam: array of const);
var
  data: PDataParam;
begin
  New(data);
  data^.refCount := 0;
  data^.data := ActParamToBuffer(AParam);
  Send(ADestActor, AAct, @data^.data);
end;

procedure TActor.Send(AAct: TAct; AParam: Pointer);
begin
  Send(Self, AAct, AParam);
end;

procedure TActor.Send(AAct: TAct; AParam: array of const);
begin
  Send(Self, AAct, AParam);
end;

end.

